package com.ruoyi.rabbitmq.service;

import com.ruoyi.rabbitmq.config.RabbitMQConfig;
import com.ruoyi.rabbitmq.domain.MessageInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;

import java.util.UUID;
import java.util.concurrent.CompletableFuture;

/**
 * 消息生产者服务
 * 
 * @author ruoyi
 */
@Service
@ConditionalOnProperty(name = "spring.rabbitmq.enabled", havingValue = "true", matchIfMissing = true)
public class MessagePublisher
{
    private static final Logger log = LoggerFactory.getLogger(MessagePublisher.class);

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 发送消息到直连交换机
     * 
     * @param content 消息内容
     * @return 消息ID
     */
    public String sendDirectMessage(Object content)
    {
        return sendMessage(content, RabbitMQConfig.DIRECT_EXCHANGE, RabbitMQConfig.DIRECT_ROUTING_KEY, "DIRECT");
    }

    /**
     * 发送消息到主题交换机
     * 
     * @param content 消息内容
     * @param routingKey 路由键
     * @return 消息ID
     */
    public String sendTopicMessage(Object content, String routingKey)
    {
        return sendMessage(content, RabbitMQConfig.TOPIC_EXCHANGE, routingKey, "TOPIC");
    }

    /**
     * 发送消息到扇形交换机
     * 
     * @param content 消息内容
     * @return 消息ID
     */
    public String sendFanoutMessage(Object content)
    {
        return sendMessage(content, RabbitMQConfig.FANOUT_EXCHANGE, "", "FANOUT");
    }

    /**
     * 发送消息到死信队列
     * 
     * @param content 消息内容
     * @return 消息ID
     */
    public String sendDeadLetterMessage(Object content)
    {
        return sendMessage(content, RabbitMQConfig.DEAD_LETTER_EXCHANGE, RabbitMQConfig.DEAD_LETTER_ROUTING_KEY, "DEAD_LETTER");
    }

    /**
     * 发送消息到重试队列
     * 
     * @param content 消息内容
     * @param delayMillis 延迟时间（毫秒）
     * @return 消息ID
     */
    public String sendRetryMessage(Object content, long delayMillis)
    {
        String messageId = UUID.randomUUID().toString();
        
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor()
        {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException
            {
                MessageProperties properties = message.getMessageProperties();
                properties.setExpiration(String.valueOf(delayMillis));
                properties.setMessageId(messageId);
                return message;
            }
        };

        try
        {
            rabbitTemplate.convertAndSend(RabbitMQConfig.RETRY_QUEUE, content, messagePostProcessor);
            log.info("消息发送到重试队列成功，消息ID: {}, 延迟时间: {}ms", messageId, delayMillis);
            return messageId;
        }
        catch (Exception e)
        {
            log.error("消息发送到重试队列失败，消息ID: {}", messageId, e);
            throw new RuntimeException("消息发送失败", e);
        }
    }

    /**
     * 通用发送消息方法
     * 
     * @param content 消息内容
     * @param exchange 交换机名称
     * @param routingKey 路由键
     * @param messageType 消息类型
     * @return 消息ID
     */
    public String sendMessage(Object content, String exchange, String routingKey, String messageType)
    {
        String messageId = UUID.randomUUID().toString();
        MessageInfo messageInfo = new MessageInfo(content, messageType, exchange, routingKey);
        messageInfo.setMessageId(messageId);

        CorrelationData correlationData = new CorrelationData(messageId);

        try
        {
            rabbitTemplate.convertAndSend(exchange, routingKey, messageInfo, correlationData);
            log.info("消息发送成功，消息ID: {}, 交换机: {}, 路由键: {}", messageId, exchange, routingKey);
            return messageId;
        }
        catch (Exception e)
        {
            log.error("消息发送失败，消息ID: {}, 交换机: {}, 路由键: {}", messageId, exchange, routingKey, e);
            throw new RuntimeException("消息发送失败", e);
        }
    }

    /**
     * 异步发送消息
     * 
     * @param content 消息内容
     * @param exchange 交换机名称
     * @param routingKey 路由键
     * @param messageType 消息类型
     * @return CompletableFuture
     */
    public CompletableFuture<String> sendMessageAsync(Object content, String exchange, String routingKey, String messageType)
    {
        return CompletableFuture.supplyAsync(() -> sendMessage(content, exchange, routingKey, messageType));
    }

    /**
     * 发送带过期时间的消息
     * 
     * @param content 消息内容
     * @param exchange 交换机名称
     * @param routingKey 路由键
     * @param messageType 消息类型
     * @param expiration 过期时间（毫秒）
     * @return 消息ID
     */
    public String sendMessageWithExpiration(Object content, String exchange, String routingKey, String messageType, long expiration)
    {
        String messageId = UUID.randomUUID().toString();
        MessageInfo messageInfo = new MessageInfo(content, messageType, exchange, routingKey);
        messageInfo.setMessageId(messageId);

        MessagePostProcessor messagePostProcessor = new MessagePostProcessor()
        {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException
            {
                MessageProperties properties = message.getMessageProperties();
                properties.setExpiration(String.valueOf(expiration));
                properties.setMessageId(messageId);
                return message;
            }
        };

        CorrelationData correlationData = new CorrelationData(messageId);

        try
        {
            rabbitTemplate.convertAndSend(exchange, routingKey, messageInfo, messagePostProcessor, correlationData);
            log.info("消息发送成功，消息ID: {}, 交换机: {}, 路由键: {}, 过期时间: {}ms", messageId, exchange, routingKey, expiration);
            return messageId;
        }
        catch (Exception e)
        {
            log.error("消息发送失败，消息ID: {}, 交换机: {}, 路由键: {}", messageId, exchange, routingKey, e);
            throw new RuntimeException("消息发送失败", e);
        }
    }

    /**
     * 发送消息到指定队列
     * 
     * @param content 消息内容
     * @param queueName 队列名称
     * @return 消息ID
     */
    public String sendMessageToQueue(Object content, String queueName)
    {
        String messageId = UUID.randomUUID().toString();
        MessageInfo messageInfo = new MessageInfo(content, "QUEUE", "", "");
        messageInfo.setMessageId(messageId);
        messageInfo.setQueueName(queueName);

        CorrelationData correlationData = new CorrelationData(messageId);

        try
        {
            rabbitTemplate.convertAndSend("", queueName, messageInfo, correlationData);
            log.info("消息发送到队列成功，消息ID: {}, 队列: {}", messageId, queueName);
            return messageId;
        }
        catch (Exception e)
        {
            log.error("消息发送到队列失败，消息ID: {}, 队列: {}", messageId, queueName, e);
            throw new RuntimeException("消息发送失败", e);
        }
    }
} 